<?php
namespace houze\rabbitmq;

use houze\rabbitmq\lib\RabbitMQPool;
use houze\rabbitmq\Connection\AMQPSwooleConnection;

class Pool extends RabbitMQPool{
    /**
     * 启动连接池
    */
    public function init() {
        if ($this->config['pool_num'] <= 0) return false;
        
        $start_time = explode(' ',microtime());
        
        $this->min = ($this->config['pool_num']>1) ? ceil($this->config['pool_num']/2) : 1;
        $this->max = $this->config['pool_num'];
            
        for ($i=0; $i<$this->max; $i++) {
            $this->connections[] = $this->create();
        }
        
        \houze\StartRecord::rabbitmq_reload($start_time);
        \houze\StartRecord::rabbitmq_monitor();
        
        return $this;
    }

    /**
     * 获取一个连接
    */
    public function pop() {
        // 重置连接
        $this->recovery();
        
        // 队列被取空
        if (!$this->connections) {
            // 新建连接入池
            $obj = $this->create();
        } else {
            $obj = array_pop($this->connections);
        }
        return $obj['db'];
    }

    /**
     * 归还一个连接
    */
    public function free($db) {
        if ($db) {
            if(count($this->connections) < $this->max){
                $obj = [
                    'last_used_time' => time(),
                    'db' => $db,
                ];
            
                $this->connections[] = $obj;
            }else{
                unset($db);
            }
            
            // 统计连接数
            $this->count($this->connections);
            return true;
        }
        return false;
    }
    
    /**
     * 重置连接
    */
    protected function recovery() {
        // 没用就回收重置连接时间
        $spare_time = $this->config['spare_time'];
        
        $list = [];
        // 阻塞循环
        while (true){
            if ($this->connections) {
                $obj = array_pop($this->connections);
                # 拿出最近一次使用时间
                $last_used_time = $obj['last_used_time'];
                # 判断回收重连超期时间
                if (time() - $last_used_time > $spare_time) {
                    array_push($list, $this->create());  
                }else{
                    array_push($list, $obj);
                }
            }else{
                break;
            }
        }
        foreach ($list as $item) {
            $this->connections[] = $item;
        }
        // 统计连接数
        $this->count($list);
        
        unset($list);
    }
    
    /**
     * 统计重置连接数
    */ 
    protected function count($arr){
        $num = count($arr);
        $path = BOX_PATH.'env'.DS.'rabbitmq_pool_num.count';
        $json = file_get_contents($path);
        $array = [];
        if ($json) {
            $array = json_decode($json, true);
        }
        $array[0] = $num;
        file_put_contents($path, json_encode($array));
       
        unset($json);
        unset($array);
        unset($path);
    }

    /**
     * 创建连接实例
     */
    protected function create() {
        $config = $this->config;

        try {
            $manager = new AMQPSwooleConnection($config['host'],$config['port'],$config['user'],$config['password'],$config['vhost']);
        } catch(\Exception $e){
            throw new \Exception("new RabbitMQ Error ".$e->getMessage());
            return false;
        }
        return [
            'last_used_time' => time(),
            'db' => $manager,
        ];
    }
}